// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <cantor/type/type.h>
#include <turbo/crypto/md5.h>

#include "cantor/parser/ast/stmt.h"

namespace cantor {
    namespace detail {
        turbo::uint128 compute_signature(std::string_view str) {
            turbo::MD5 hash;
            hash.update(str);
            auto r = hash.result();
            return turbo::make_uint128(r.high(), r.low());
        }
    }

    turbo::Status check_unique(const std::vector<std::string> &values) {
        std::unordered_set<std::string> values_set;
        for (const auto &value: values) {
            if (values_set.find(value) != values_set.end()) {
                return turbo::already_exists_error(value);
            }
            values_set.insert(value);
        }
        return turbo::OkStatus();
    }

    namespace {
        std::string names_and_types_to_string(
            const std::vector<std::string> &names,
            const std::vector<TypePtr> &types) {
            std::stringstream ss;
            ss << "[names: {";
            if (!names.empty()) {
                for (const auto &name: names) {
                    ss << "'" << name << "', ";
                }
                ss.seekp(-2, std::ios_base::cur);
            } else {
                ss << " ";
            }
            ss << "}, types: {";
            if (!types.empty()) {
                for (const auto &type: types) {
                    ss << (type ? type->full_name : "NULL") << ", ";
                }
                ss.seekp(-2, std::ios_base::cur);
            } else {
                ss << " ";
            }
            ss << "}]";
            return ss.str();
        }
    }


    ///////////////////////////////////////////////////////////////////////////////////
    /// ARRAY
    ///////////////////////////////////////////////////////////////////////////////////


    std::string ArrayType::make_bound(const std::vector<int64_t> &dims) {
        std::stringstream ss;
        for (const auto &dim: dims) {
            ss << "[";
            if (dim > 0) {
                ss << dim;
            }
            ss << "]";
        }
        return ss.str();
    }

    bool ArrayType::check_fixed(const std::vector<int64_t> &bounds) {
        for (const auto &dim: bounds) {
            if (dim <= 0) {
                return false;
            }
        }
        return true;
    }

    turbo::uint128 ArrayType::init_signature(const TypePtr &child, const std::vector<int64_t> &dims) {
        std::stringstream ss;
        for (const auto &dim: dims) {
            ss << "[" << dim << "]";
        }
        auto l0_str = turbo::str_cat("L0:", TYPE_NAME, "{ dims:", ss.str(), "fixed:", check_fixed(dims), "}");
        auto l1_str = turbo::str_cat("L1:", "h:", uint128_high64(child->signature), "l:",
                                     uint128_low64(child->signature), "}");
        return detail::compute_signature(l0_str + l1_str);
    }

    const std::shared_ptr<const Type> &ArrayType::child_at(uint32_t idx) const {
        KCHECK_EQ(idx, 0) << "Array type should have only one child";
        return child_;
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// MAP
    ///////////////////////////////////////////////////////////////////////////////////

    turbo::uint128 MapType::init_signature(const TypePtr &keyType, const TypePtr &valueType) {
        auto l0_str = turbo::str_cat("L0:{", TYPE_NAME, "}");
        auto key_str = turbo::str_cat("L1:{key:{", "h:", uint128_high64(keyType->signature), "l:",
                                      uint128_low64(keyType->signature), "}}");
        auto value_str = turbo::str_cat("L1:{value:{", "h:", uint128_high64(valueType->signature), "l:",
                                        uint128_low64(valueType->signature), "}}");
        return detail::compute_signature(l0_str + key_str + value_str);
    }

    const TypePtr &MapType::child_at(uint32_t idx) const {
        KCHECK(idx <2) << "Map type should have only two children. Tried to access child " << idx;
        if (idx == 0) {
            return key_type();
        }
        return value_type();
    }

    const char *MapType::name_of(uint32_t idx) const {
        KCHECK(idx <2) << "Map type should have only two children. Tried to access child " << idx;
        if (idx == 0) {
            return "key";
        }
        return "value";
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// SET
    ///////////////////////////////////////////////////////////////////////////////////

    turbo::uint128 SetType::init_signature(const TypePtr &keyType) {
        auto l0_str = turbo::str_cat("L0:{", TYPE_NAME, "}");
        auto key_str = turbo::str_cat("L1:{key:{", "h:", uint128_high64(keyType->signature), "l:",
                                      uint128_low64(keyType->signature), "}}");
        return detail::compute_signature(l0_str + key_str);
    }

    const TypePtr &SetType::child_at(uint32_t idx) const {
        KCHECK(idx == 0) << "SET type should have only two children. Tried to access child " << idx;
        return keyType();
    }

    const char *SetType::name_of(uint32_t idx) const {
        KCHECK(idx == 0) << "SET type should have only two children. Tried to access child " << idx;
        return "key";
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// STRUCT
    ///////////////////////////////////////////////////////////////////////////////////
    ///
    StructType::StructType(std::vector<std::string> &&names, std::vector<TypePtr> &&types)
        : TypeBase<pb::TypeKind::TYPE_KIND_STRUCT>(std::all_of(
                                                   types.cbegin(), types.cend(), [](const auto &child) {
                                                       return child->is_orderable;
                                                   }), std::all_of(
                                                   types.cbegin(), types.cend(), [](const auto &child) {
                                                       return child->is_comparable;
                                                   }), init_signature(names, types), make_full_name(names, types)), names_{std::move(names)}, children_{std::move(types)} {
        KCHECK_EQ(
            names_.size(),
            children_.size()) <<
            "Mismatch names/types sizes: ",
            names_and_types_to_string(names_, children_);
        for (auto &child: children_) {
            KCHECK_NE(child, nullptr) <<
                "Child types cannot be null: {}",
                names_and_types_to_string(names_, children_);
        }
    }

    turbo::uint128 StructType::init_signature(const std::vector<std::string> &names,
            const std::vector<std::shared_ptr<const Type> > &types) {
        auto ns = turbo::str_join(names, ",");
        auto str = turbo::str_cat("L0:STRUCT{fields:",names.size(), "names:[", ns, "]}");
        std::ostringstream ss;
        ss<<"l1:{";
        for (size_t i = 0; i < types.size(); i++) {
            ss << i<<":"<<types[i]->signature;
        }
        ss<<"}";
        return detail::compute_signature(str + ss.str());
    }

    namespace {
        template<typename T>
        std::string make_field_not_found_error_eessage(
            const T &name,
            const std::vector<std::string> &availableNames) {
            std::stringstream errorMessage;
            errorMessage << "Field not found: " << name << ". Available fields are: ";
            for (auto i = 0; i < availableNames.size(); ++i) {
                if (i > 0) {
                    errorMessage << ", ";
                }
                errorMessage << availableNames[i];
            }
            errorMessage << ".";
            return errorMessage.str();
        }
    } // namespace

    const TypePtr &StructType::findChild(std::string_view name) const {
        for (uint32_t i = 0; i < names_.size(); ++i) {
            if (names_.at(i) == name) {
                return children_.at(i);
            }
        }
        KCHECK(false) << make_field_not_found_error_eessage(name, names_);
        static TypePtr ptr;
        return ptr;
    }

    bool StructType::contains_child(std::string_view name) const {
        return std::find(names_.begin(), names_.end(), name) != names_.end();
    }

    uint32_t StructType::get_child_idx(std::string_view name) const {
        auto index = get_child_idx_if_exists(name);
        if (!index.has_value()) {
            KCHECK(false) << make_field_not_found_error_eessage(name, names_);
        }
        return index.value();
    }

    std::optional<uint32_t> StructType::get_child_idx_if_exists(
        std::string_view name) const {
        for (uint32_t i = 0; i < names_.size(); i++) {
            if (names_.at(i) == name) {
                return i;
            }
        }
        return std::nullopt;
    }

    void StructType::print_children(std::stringstream &ss ,  const std::vector<std::string> &names,
             const std::vector<std::shared_ptr<const Type> > &types, std::string_view delimiter){
        bool any = false;
        for (size_t i = 0; i < types.size(); ++i) {
            if (any) {
                ss << delimiter;
            }
            const auto &name = names.at(i);
            ss << name;
            ss << ':' << types.at(i)->full_name;
            any = true;
        }
    }

    std::string StructType::make_full_name(const std::vector<std::string> &names,
             const std::vector<std::shared_ptr<const Type> > &types) {
        std::stringstream ss;
        ss << (TypeTraits<pb::TypeKind::TYPE_KIND_STRUCT>::name) << "<";
        print_children(ss, names, types);
        ss << ">";
        return ss.str();
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// TUPLE
    ///////////////////////////////////////////////////////////////////////////////////
    ///
    TupleType::TupleType(std::vector<TypePtr> &&types)
        : TypeBase<pb::TypeKind::TYPE_KIND_TUPLE>(std::all_of(
                                                  types.cbegin(), types.cend(), [](const auto &child) {
                                                      return child->is_orderable;
                                                  }), std::all_of(
                                                  types.cbegin(), types.cend(), [](const auto &child) {
                                                      return child->is_comparable;
                                                  }), init_signature(types), make_full_name(types)), children_{std::move(types)} {
        for (auto &child: children_) {
            KCHECK_NE(child, nullptr) <<
                "Child types cannot be null:";
        }
    }

    TupleType::~TupleType() = default;

    turbo::uint128 TupleType::init_signature(const std::vector<std::shared_ptr<const Type> > &types) {
        auto str = turbo::str_cat("L0:UNION{fields:",types.size(), "]}");
        std::ostringstream ss;
        ss<<"l1:{";
        for (size_t i = 0; i < types.size(); i++) {
            ss << i<<":"<<types[i]->signature;
        }
        ss<<"}";
        return detail::compute_signature(str + ss.str());
    }

    void TupleType::print_children(std::stringstream &ss, const std::vector<std::shared_ptr<const Type> > &types, std::string_view delimiter){
        bool any = false;
        for (size_t i = 0; i < types.size(); ++i) {
            if (any) {
                ss << delimiter;
            }
            ss << i;
            ss << ':' << types.at(i)->full_name;
            any = true;
        }
    }

    std::string TupleType::make_full_name(const std::vector<std::shared_ptr<const Type> > &types) {
        std::stringstream ss;
        ss << (TypeTraits<pb::TypeKind::TYPE_KIND_TUPLE>::name) << "<";
        print_children(ss, types);
        ss << ">";
        return ss.str();
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// UNION
    ///////////////////////////////////////////////////////////////////////////////////
    ///
    UnionType::UnionType(std::vector<std::string> &&names, std::vector<TypePtr> &&types)
        : TypeBase<pb::TypeKind::TYPE_KIND_UNION>(std::all_of(
                                                  types.cbegin(), types.cend(), [](const auto &child) {
                                                      return child->is_orderable;
                                                  }), std::all_of(
                                                  types.cbegin(), types.cend(), [](const auto &child) {
                                                      return child->is_comparable;
                                                  }), init_signature(names, types),make_full_name(names, types)), names_{std::move(names)}, children_{std::move(types)} {
        KCHECK_EQ(
            names_.size(),
            children_.size()) <<
            "Mismatch names/types sizes: ",
            names_and_types_to_string(names_, children_);
        for (auto &child: children_) {
            KCHECK_NE(child, nullptr) <<
                "Child types cannot be null: {}",
                names_and_types_to_string(names_, children_);
        }
    }

    turbo::uint128 UnionType::init_signature(const std::vector<std::string> &names,
            const std::vector<std::shared_ptr<const Type> > &types) {
        auto ns = turbo::str_join(names, ",");
        auto str = turbo::str_cat("L0:UNION{fields:",names.size(), "names:[", ns, "]}");
        std::ostringstream ss;
        ss<<"l1:{";
        for (size_t i = 0; i < types.size(); i++) {
            ss << i<<":"<<types[i]->signature;
        }
        ss<<"}";
        return detail::compute_signature(str + ss.str());
    }

    UnionType::~UnionType() = default;

    const TypePtr &UnionType::findChild(std::string_view name) const {
        for (uint32_t i = 0; i < names_.size(); ++i) {
            if (names_.at(i) == name) {
                return children_.at(i);
            }
        }
        KCHECK(false) << make_field_not_found_error_eessage(name, names_);
        static TypePtr ptr;
        return ptr;
    }


    bool UnionType::contains_child(std::string_view name) const {
        return std::find(names_.begin(), names_.end(), name) != names_.end();
    }

    uint32_t UnionType::get_child_idx(std::string_view name) const {
        auto index = get_child_idx_if_exists(name);
        if (!index.has_value()) {
            KCHECK(false) << make_field_not_found_error_eessage(name, names_);
        }
        return index.value();
    }

    std::optional<uint32_t> UnionType::get_child_idx_if_exists(
        std::string_view name) const {
        for (uint32_t i = 0; i < names_.size(); i++) {
            if (names_.at(i) == name) {
                return i;
            }
        }
        return std::nullopt;
    }

    void UnionType::print_children(std::stringstream &ss, const std::vector<std::string> &names,
                                             const std::vector<std::shared_ptr<const Type> > &types, std::string_view delimiter) {
        bool any = false;
        for (size_t i = 0; i < types.size(); ++i) {
            if (any) {
                ss << delimiter;
            }
            const auto &name = names.at(i);
            ss << name;
            ss << ':' << types.at(i)->full_name;
            any = true;
        }
    }

    std::string UnionType::make_full_name(const std::vector<std::string> &names,
                                             const std::vector<std::shared_ptr<const Type> > &types) {
        std::stringstream ss;
        ss << (TypeTraits<pb::TypeKind::TYPE_KIND_UNION>::name) << "<";
        print_children(ss, names, types);
        ss << ">";
        return ss.str();
    }

    bool Type::kind_equals(const TypePtr &other) const {
        // recursive kind match (ignores names)
        if (this->kind() != other->kind()) {
            return false;
        }
        if (this->size() != other->size()) {
            return false;
        }
        for (size_t i = 0; i < this->size(); ++i) {
            if (!this->child_at(i)->kind_equals(other->child_at(i))) {
                return false;
            }
        }
        return true;
    }


    size_t Type::hash_kind() const {
        size_t hash = (int32_t) kind() + 1;
        for (auto &child: *this) {
            hash = hash * 31 + child->hash_kind();
        }
        return hash;
    }

    turbo::Result<uint32_t> EnumType::construct_enum(const std::string &em) const {
        auto up_name = turbo::str_to_upper(em);
        auto it = _data.name_to_value.find(up_name);
        if (it == _data.name_to_value.end()) {
            return turbo::not_found_error("not found ", up_name);
        }
        return it->second;
    }

    turbo::Status EnumType::construct_enum(uint32_t ev) const {
        auto it = _data.value_to_name.find(ev);
        if (it == _data.value_to_name.end()) {
            return turbo::not_found_error("not found ", ev);
        }
        return turbo::OkStatus();
    }

    turbo::Result<std::string> EnumType::resolve_enum(uint32_t ev) const {
        auto it = _data.value_to_name.find(ev);
        if (it == _data.value_to_name.end()) {
            return turbo::not_found_error("not found ", ev);
        }
        KCHECK(!it->second.empty());
        return it->second[0];
    }

    turbo::Result<std::string> EnumType::resolve_enum(uint32_t ev, resolve_enum_handler handler) const {
        if (!handler) {
            return turbo::invalid_argument_error("handler is null");
        }
        auto it = _data.value_to_name.find(ev);
        if (it == _data.value_to_name.end()) {
            return turbo::not_found_error("not found ", ev);
        }
        return handler(it->second);
    }

    ///////////////////////////////////////////////////////////////////////////////////
    /// CREATORS
    ///////////////////////////////////////////////////////////////////////////////////
    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_BOOLEAN>> BOOLEAN() {
        return NumericType<pb::TypeKind::TYPE_KIND_BOOLEAN>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_INT8>> INT8() {
        return NumericType<pb::TypeKind::TYPE_KIND_INT8>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_INT16>> INT16() {
        return NumericType<pb::TypeKind::TYPE_KIND_INT16>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_INT32>> INT32() {
        return NumericType<pb::TypeKind::TYPE_KIND_INT32>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_INT64>> INT64() {
        return NumericType<pb::TypeKind::TYPE_KIND_INT64>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_INT128>> INT128() {
        return NumericType<pb::TypeKind::TYPE_KIND_INT128>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_UINT8>> UINT8() {
        return NumericType<pb::TypeKind::TYPE_KIND_UINT8>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_UINT16>> UINT16() {
        return NumericType<pb::TypeKind::TYPE_KIND_UINT16>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_UINT32>> UINT32() {
        return NumericType<pb::TypeKind::TYPE_KIND_UINT32>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_UINT64>> UINT64() {
        return NumericType<pb::TypeKind::TYPE_KIND_UINT64>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_UINT128>> UINT128() {
        return NumericType<pb::TypeKind::TYPE_KIND_UINT128>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_FP8>> FP8() {
        return NumericType<pb::TypeKind::TYPE_KIND_FP8>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_FP16>> FP16() {
        return NumericType<pb::TypeKind::TYPE_KIND_FP16>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_FP32>> FP32() {
        return NumericType<pb::TypeKind::TYPE_KIND_FP32>::create();
    }

    std::shared_ptr<const NumericType<pb::TypeKind::TYPE_KIND_FP64>> FP64() {
        return NumericType<pb::TypeKind::TYPE_KIND_FP64>::create();
    }

    std::shared_ptr<const StringType> STRING(uint32_t max) {
        return std::make_shared<StringType>(false, max);
    }

    std::shared_ptr<const StringType> FIXED_STRING(uint32_t max) {
        return std::make_shared<StringType>(true, max);
    }

    std::shared_ptr<const BinaryType> BINARY(uint32_t max) {
        return std::make_shared<BinaryType>(false, max);
    }

    std::shared_ptr<const BinaryType> FIXED_BINARY(uint32_t max) {
        return std::make_shared<BinaryType>(true, max);
    }

    turbo::Result<std::shared_ptr<const TimestampType> > TIMESTAMP(pb::TimePrecision precision, const std::string &tz) {
        if (!is_valid_timestamp_precision(precision)) {
            return turbo::invalid_argument_error("invalid precision");
        }
        return std::make_shared<TimestampType>(precision, tz);
    }

    std::shared_ptr<const TimestampType> NANO_TIMESTAMP(const std::string &tz) {
        return std::make_shared<TimestampType>(pb::TimePrecision::TSP_NANO, tz);
    }

    std::shared_ptr<const TimestampType> MICRO_TIMESTAMP(const std::string &tz) {
        return std::make_shared<TimestampType>(pb::TimePrecision::TSP_MICRO, tz);
    }

    std::shared_ptr<const TimestampType> MILLI_TIMESTAMP(const std::string &tz) {
        return std::make_shared<TimestampType>(pb::TimePrecision::TSP_MILLI, tz);
    }

    std::shared_ptr<const TimestampType> SECOND_TIMESTAMP(const std::string &tz) {
        return std::make_shared<TimestampType>(pb::TimePrecision::TSP_SECONDS, tz);
    }

    turbo::Result<std::shared_ptr<const DurationType> > DURATION(pb::TimePrecision precision) {
        if (!is_valid_timestamp_precision(precision)) {
            return turbo::invalid_argument_error("invalid precision");
        }
        switch (precision) {
            case pb::TimePrecision::TSP_NANO:
                return NANO_DURATION();
            case pb::TimePrecision::TSP_MICRO:
                return MICRO_DURATION();
            case pb::TimePrecision::TSP_MILLI:
                return MILLI_DURATION();
            case pb::TimePrecision::TSP_SECONDS:
                return SECOND_DURATION();
        }
        TURBO_UNREACHABLE();
    }

    std::shared_ptr<const DurationType> NANO_DURATION() {
        static std::shared_ptr<DurationType> duration = std::make_shared<DurationType>(pb::TimePrecision::TSP_NANO);
        return duration;
    }

    std::shared_ptr<const DurationType> MICRO_DURATION() {
        static std::shared_ptr<DurationType> duration = std::make_shared<DurationType>(pb::TimePrecision::TSP_MICRO);
        return duration;
    }

    std::shared_ptr<const DurationType> MILLI_DURATION() {
        static std::shared_ptr<DurationType> duration = std::make_shared<DurationType>(pb::TimePrecision::TSP_MILLI);
        return duration;
    }

    std::shared_ptr<const DurationType> SECOND_DURATION() {
        static std::shared_ptr<DurationType> duration = std::make_shared<DurationType>(pb::TimePrecision::TSP_SECONDS);
        return duration;
    }


    /// DATETIME
    turbo::Result<std::shared_ptr<const DateTimeType> > DATETIME(pb::DateTimePrecision precision, const std::string &tz) {
        if (!is_valid_date_precision(precision)) {
            return turbo::invalid_argument_error("invalid precision");
        }
        return std::make_shared<DateTimeType>(precision, tz);
    }

    std::shared_ptr<const DateTimeType> NANO_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_NANO, tz);
    }

    std::shared_ptr<const DateTimeType> MICRO_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_MICRO, tz);
    }

    std::shared_ptr<const DateTimeType> MILLI_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_MILLI, tz);
    }

    std::shared_ptr<const DateTimeType> SECOND_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_SECOND, tz);
    }

    std::shared_ptr<const DateTimeType> MINUTE_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_MINUTE, tz);
    }

    std::shared_ptr<const DateTimeType> HOUR_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_HOUR, tz);
    }

    std::shared_ptr<const DateTimeType> DAY_DATETIME(const std::string &tz) {
        return std::make_shared<DateTimeType>(pb::DateTimePrecision::DTP_DAY, tz);
    }

    /// INTERVAL
    turbo::Result<std::shared_ptr<const IntervalType> > INTERVAL(pb::DateTimePrecision precision) {
        if (!is_valid_date_precision(precision)) {
            return turbo::invalid_argument_error("invalid date precision");
        }
        switch (precision) {
            case pb::DateTimePrecision::DTP_DAY:
                return DAY_INTERVAL();
            case pb::DateTimePrecision::DTP_HOUR:
                return HOUR_INTERVAL();
            case pb::DateTimePrecision::DTP_MINUTE:
                return MINUTE_INTERVAL();
            case pb::DateTimePrecision::DTP_SECOND:
                return SECOND_INTERVAL();
            case pb::DateTimePrecision::DTP_MILLI:
                return MILLI_INTERVAL();
            case pb::DateTimePrecision::DTP_MICRO:
                return MICRO_INTERVAL();
            case pb::DateTimePrecision::DTP_NANO:
                return NANO_INTERVAL();
        }
        TURBO_UNREACHABLE();
    }

    std::shared_ptr<const IntervalType> NANO_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_NANO);
        return interval;
    }

    std::shared_ptr<const IntervalType> MICRO_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_MICRO);
        return interval;
    }

    std::shared_ptr<const IntervalType> MILLI_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_MILLI);
        return interval;
    }

    std::shared_ptr<const IntervalType> SECOND_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_SECOND);
        return interval;
    }

    std::shared_ptr<const IntervalType> MINUTE_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_MINUTE);
        return interval;
    }

    std::shared_ptr<const IntervalType> HOUR_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_HOUR);
        return interval;
    }

    std::shared_ptr<const IntervalType> DAY_INTERVAL() {
        static auto interval = std::make_shared<IntervalType>(pb::DateTimePrecision::DTP_DAY);
        return interval;
    }

    turbo::Result<std::shared_ptr<const DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL64>> > DECIMAL64(
        const uint8_t precision, const uint8_t scale) {
        TURBO_RETURN_NOT_OK(DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL64>::check_parameters(precision, scale));
        return std::make_shared<DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL64> >(precision, scale);
    }

    turbo::Result<std::shared_ptr<const DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL128>> > DECIMAL128(
        const uint8_t precision, const uint8_t scale) {
        TURBO_RETURN_NOT_OK(DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL128>::check_parameters(precision, scale));
        return std::make_shared<DecimalType<pb::TypeKind::TYPE_KIND_DECIMAL128> >(precision, scale);
    }

    turbo::Result<std::shared_ptr<const EnumType> > ENUM(EnumData data) {
        TURBO_RETURN_NOT_OK(data.status);
        return std::make_shared<EnumType>(std::move(data));
    }

    /// ARRAY
    turbo::Result<std::shared_ptr<const ArrayType> > ARRAY(TypePtr ptr, std::vector<int64_t> dims) {
        if (dims.empty()) {
            return turbo::invalid_argument_error("empty array");
        }
        return std::make_shared<ArrayType>(std::move(ptr), std::move(dims));
    }

    turbo::Result<std::shared_ptr<const ArrayType> > ARRAY_UNLIMITED(TypePtr ptr, uint32_t dim) {
        if (dim == 0) {
            return turbo::invalid_argument_error("empty array");
        }
        std::vector<int64_t> dims;
        dims.reserve(dim);
        for (uint32_t i = 0; i < dim; i++) {
            dims.push_back(-1);
        }
        return ARRAY(std::move(ptr), dims);
    }

    turbo::Result<std::shared_ptr<const ArrayType> > ARRAY_WITH_BOUND(TypePtr ptr, int64_t dim) {
        std::vector<int64_t> dims;
        dims.push_back(dim);
        return ARRAY(std::move(ptr), dims);
    }

    turbo::Result<std::shared_ptr<const ArrayType> > ARRAY_WITH_BOUND(TypePtr ptr, int64_t dim0, int64_t dim1) {
        std::vector<int64_t> dims;
        dims.push_back(dim0);
        dims.push_back(dim1);
        return ARRAY(std::move(ptr), dims);
    }

    turbo::Result<std::shared_ptr<const ArrayType> > ARRAY_WITH_BOUND(TypePtr ptr, int64_t dim0, int64_t dim1,
                                                                      int64_t dim2) {
        std::vector<int64_t> dims;
        dims.push_back(dim0);
        dims.push_back(dim1);
        return ARRAY(std::move(ptr), dims);
    }

    turbo::Result<std::shared_ptr<const MapType> > MAP(TypePtr key, TypePtr value) {
        if (key == nullptr) {
            return turbo::invalid_argument_error("empty key type");
        }
        if (value == nullptr) {
            return turbo::invalid_argument_error("empty value type");
        }
        if (!key->is_comparable) {
            return turbo::invalid_argument_error("key is not comparable");
        }
        return std::make_shared<MapType>(std::move(key), std::move(value));
    }


    turbo::Result<std::shared_ptr<const SetType> > SET(TypePtr key) {
        if (key == nullptr) {
            return turbo::invalid_argument_error("empty key type");
        }
        if (!key->is_comparable) {
            return turbo::invalid_argument_error("key is not comparable");
        }
        return std::make_shared<SetType>(std::move(key));
    }


    turbo::Result<std::shared_ptr<const StructType> > STRUCT(std::vector<std::string> &&names,
                                                             std::vector<std::shared_ptr<const Type> > &&types) {
        if (names.empty()) {
            return turbo::invalid_argument_error("empty names array");
        }
        auto rs = check_unique(names);
        if (!rs.ok()) {
            return turbo::invalid_argument_error("duplicate field name: " + rs.message());
        }
        if (types.empty()) {
            return turbo::invalid_argument_error("empty types array");
        }
        if (names.size() != types.size()) {
            return turbo::invalid_argument_error("names and types array must have the same size");
        }

        return std::make_shared<StructType>(std::move(names), std::move(types));
    }

    turbo::Result<std::shared_ptr<const TupleType> > TUPLE(std::vector<std::shared_ptr<const Type> > &&types) {
        if (types.empty()) {
            return turbo::invalid_argument_error("empty types array");
        }
        return std::make_shared<TupleType>(std::move(types));
    }

    turbo::Result<std::shared_ptr<const UnionType> > UNION(std::vector<std::string> &&names,
                                                           std::vector<std::shared_ptr<const Type> > &&types) {
        if (names.empty()) {
            return turbo::invalid_argument_error("empty names array");
        }
        auto rs = check_unique(names);
        if (!rs.ok()) {
            return turbo::invalid_argument_error("duplicate field name: " + rs.message());
        }
        if (types.empty()) {
            return turbo::invalid_argument_error("empty types array");
        }
        if (names.size() != types.size()) {
            return turbo::invalid_argument_error("names and types array must have the same size");
        }
        return std::make_shared<UnionType>(std::move(names), std::move(types));
    }

    turbo::Result<std::shared_ptr<const UdfType> > UDF(const std::string &uri) {
        if (uri.empty()) {
            return turbo::invalid_argument_error("empty uri");
        }
        return UdfType::create(uri);
    }

    turbo::Result<std::shared_ptr<const UdfType> > UDF(const std::string &uri, const TypePtr &type) {
        if (uri.empty()) {
            return turbo::invalid_argument_error("empty uri");
        }
        if (!type) {
            return UdfType::create(uri);
        }
        return std::make_shared<UdfType>(uri, type);
    }

    turbo::Result<std::shared_ptr<const JsonType> > JSON(const std::string &schema) {
        return std::make_shared<JsonType>(schema);
    }

    turbo::Result<std::shared_ptr<const ProtoType> > PROTO(const std::string &uri) {
        if (uri.empty()) {
            return turbo::invalid_argument_error("empty uri");
        }
        return std::make_shared<ProtoType>(uri);
    }

} // namespace cantor
